package com.yanggu.market_analysis

import cn.hutool.core.date.DateUtil
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.util.Date

/**
 * 页面广告按照省份划分的点击量的统计
 * 设置事件时间滑动窗口, 窗口大小为1h, 滑动步长为5s
 */
object AdStatisticsByGeo {

  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)

    environment
      .readTextFile(getClass.getResource("/AdClickLog.csv").getPath)
      .map(data => {
        val arr = data.split(",")
        AdClickLog(arr(0).toLong, arr(1).toLong, arr(2), arr(3), arr(4).toLong)
      })
      .assignAscendingTimestamps(_.timestamp * 1000)
      .keyBy(_.province)
      .window(SlidingEventTimeWindows.of(Time.hours(1L), Time.seconds(5L)))
      .aggregate(new CountAgg, new CountResult)
      .print("result")

    environment.execute("AdStatisticsByGeo Job")

  }

}

case class AdClickLog(userId: Long, adId: Long, province: String, city: String, timestamp: Long)

case class CountByProvince(windowEnd: String, province: String, count: Long)

class CountAgg extends AggregateFunction[AdClickLog, Long, Long] {

  override def createAccumulator(): Long = 0L

  override def add(value: AdClickLog, accumulator: Long): Long = accumulator + 1L

  override def getResult(accumulator: Long): Long = accumulator

  override def merge(a: Long, b: Long): Long = a + b

}

class CountResult extends ProcessWindowFunction[Long, CountByProvince, String, TimeWindow] {

  override def process(key: String, context: Context, elements: Iterable[Long], out: Collector[CountByProvince]): Unit = {
    out.collect(CountByProvince(DateUtil.formatDateTime(new Date(context.window.getEnd)), key, elements.head))
  }

}
